package com.holly.unit.kafka.impl;

import cn.hutool.core.util.StrUtil;
import com.holly.unit.kafka.KafkaMetaApi;
import com.holly.unit.kafka.config.KafkaConfig;
import com.holly.unit.kafka.exception.KafkaException;
import com.holly.unit.kafka.exception.enums.KafkaExceptionEnum;
import com.holly.unit.kafka.model.Ktopic;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.TopicPartitionInfo;

import java.util.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

@Slf4j
public class KafkaMetaApiImpl implements KafkaMetaApi{


	private AdminClient adminClient;

	private static volatile String  bootstrapServers = null;
	public KafkaMetaApiImpl(KafkaConfig _kafkaConfig) {
		super();
		if(bootstrapServers == null){
			synchronized (this.getClass()) {
				if(bootstrapServers == null)
				{
					bootstrapServers = _kafkaConfig.getBootstrapServers() ;
				}
			}
		}
	}
	/**
	 * 新增topic，支持批量
	 */
	public void createTopic(Collection<Ktopic> newTopics) throws KafkaException {
		try {
			List<NewTopic> newTopicList = new ArrayList<NewTopic>();
			for(Ktopic topic : newTopics)
			{
				newTopicList.add(new NewTopic(topic.getName(), topic.getNumPartitions(), (short)topic.getReplicationFactor()));
			}
			CreateTopicsResult topicResult = getAdminClient().createTopics(newTopicList);
			topicResult.all().get();
		} catch (Exception e) {
			log.error("新增topic失败：{}",e);
			String format = StrUtil.format(KafkaExceptionEnum.KAFKA_MQ_METADATA_ERROR.getUserTip(), "新增topic失败",e.getMessage());
			throw new KafkaException(KafkaExceptionEnum.KAFKA_MQ_METADATA_ERROR.getErrorCode(), format);
		}

	}
	private synchronized AdminClient getAdminClient() {
		if(adminClient == null)
		{
			Map<String, Object> props = new HashMap<>(1);
			props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
			adminClient = KafkaAdminClient.create(props);
		}
		return adminClient;
	}

	/**
	 * 删除topic，支持批量
	 */
	public void deleteTopic(Collection<String> topics) throws KafkaException {
		try {
			DeleteTopicsResult topicResult = getAdminClient().deleteTopics(topics);
			topicResult.all().get();
		} catch (Exception e) {
			log.error("删除topic失败：{}",e);
			String format = StrUtil.format(KafkaExceptionEnum.KAFKA_MQ_METADATA_ERROR.getUserTip(), "删除topic失败",e.getMessage());
			throw new KafkaException(KafkaExceptionEnum.KAFKA_MQ_METADATA_ERROR.getErrorCode(), format);
		}
	}

	/**
	 * 获取指定topic的信息
	 */
	public String getTopicInfo(Collection<String> topics) throws KafkaException{
		AtomicReference<String> info = new AtomicReference<>("");
		try {
			getAdminClient().describeTopics(topics).all().get().forEach((topic, description) -> {
				for (TopicPartitionInfo partition : description.partitions()) {
					info.set(info + partition.toString() + "\n");
				}
			});
		} catch (Exception e) {
			log.error("获取topic失败：{}",e);
			String format = StrUtil.format(KafkaExceptionEnum.KAFKA_MQ_METADATA_ERROR.getUserTip(), "获取topic详情失败");
			throw new KafkaException(KafkaExceptionEnum.KAFKA_MQ_METADATA_ERROR.getErrorCode(), format);
		}
		return info.get();
	}

	/**
	 * 获取全部topic
	 */
	public List<String> getAllTopic() throws KafkaException {
		List<String> allTopic = new ArrayList<>();
		try {
			allTopic =  getAdminClient().listTopics().listings().get().stream().map(TopicListing::name).collect(Collectors.toList());
		} catch (Exception e ) {
			log.error("获取全部topic失败：{}",e);
			String format = StrUtil.format(KafkaExceptionEnum.KAFKA_MQ_METADATA_ERROR.getUserTip(), "获取全部topic失败");
			throw new KafkaException(KafkaExceptionEnum.KAFKA_MQ_METADATA_ERROR.getErrorCode(), format);
		}
		return allTopic;
	}
	//	 /**
	//     * 获取当前topic下的全部分区的偏移量信息
	//     *
	//     * @param properties 配置信息
	//     * @param partitions Collection<TopicPartition> partitions
	//     * @return {partition:offset}
	//	 * @throws ExecutionException 
	//	 * @throws InterruptedException 
	//     */
	//    public  Map<String, Long> getPartitionsOffset(Collection<String> topics) throws KafkaMetaException, InterruptedException, ExecutionException {
	//
	//    	Map<String, Object> props = new HashMap<>(1);
	//		props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,kafkaConfig.getBootstrapServers());
	//		
	//		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	//		props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
	//		AdminClient adminClient = KafkaAdminClient.create(props);
	//		
	//		List<TopicPartition> partitionList = new ArrayList<TopicPartition>();
	//		adminClient.describeTopics(topics).all().get().forEach((topic, description) -> {
	//			for (TopicPartitionInfo partition : description.partitions()) {
	//				TopicPartition tp = new TopicPartition(topic,partition.partition());
	//				partitionList.add(tp);
	//			}
	//		});
	//        @SuppressWarnings("rawtypes")
	//		KafkaConsumer consumer = new KafkaConsumer(props);
	//        try {
	//            Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitionList);
	//            Iterator<Map.Entry<TopicPartition, Long>> iterator = endOffsets.entrySet().iterator();
	//            while (iterator.hasNext()) {
	//                Map.Entry<TopicPartition, Long> entry = iterator.next();
	//                entry.getKey().partition()
	//                System.out.println(entry.getKey() + ": " + entry.getValue());
	//            }
	//        } catch (Exception e) {
	//            throw new KafkaMetaException(e);
	//        }finally {
	//            consumer.close();
	//        }
	//    	return null;
	//    }
}
